package com.zxymq.amqp.config;


import com.zxymq.amqp.constants.MQConstants;
import com.zxymq.amqp.support.DefaultJackson2JsonMessageConverter;
import com.zxymq.amqp.support.Gson2JsonMessageConverter;
import com.zxymq.amqp.support.RabbitMetaMessage;
import com.zxymq.amqp.util.json.JsonMapper;
import com.zxymq.amqp.util.redis.RedisUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

/**
 * RabbitTemplate配置工厂类
 *
 * @author:
 * @date :2018/1/11
 */
@Configuration
@ComponentScan
public class RabbitTemplateConfig {
    private Logger logger = LoggerFactory.getLogger(RabbitTemplateConfig.class);
    @Autowired
    private RedisUtils redisTemplate;

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter());
        // mandatory 必须设置为true，ReturnCallback才会调用
        rabbitTemplate.setMandatory(true);
        // 消息发送到RabbitMQ交换器后接收ack回调
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            logger.debug("confirm回调，ack={} correlationData={} cause={}", ack, correlationData, cause);
            String cacheKey = correlationData.getId();
            RabbitMetaMessage metaMessage = redisTemplate.getMapField(MQConstants.MQ_PRODUCER_RETRY_KEY, cacheKey, RabbitMetaMessage.class);
            // 只要消息能投入正确的交换器中，ack就为true
            if (ack) {
                if (!metaMessage.isReturnCallback()) {
                    logger.info("消息已正确投递到队列，correlationData:{}", correlationData);
                    // 清除重发缓存
                    redisTemplate.delMapField(MQConstants.MQ_PRODUCER_RETRY_KEY, cacheKey);
                } else {
                    logger.warn("交换机投机消息至队列失败，correlationData:{}", correlationData);
                }
            } else {
                logger.error("消息投递至交换机失败，correlationData:{}，原因：{}", correlationData, cause);
                if (!metaMessage.isAutoTrigger()) {
                    metaMessage.setAutoTrigger(true);
                    redisTemplate.addMap(MQConstants.MQ_PRODUCER_RETRY_KEY, cacheKey, metaMessage);
                }
            }
        });

        //消息发送到RabbitMQ交换器，但无相应Exchange时的回调
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            String cacheKey = message.getMessageProperties().getMessageId();

            logger.error("return回调，没有找到任何匹配的队列！message id:{},replyCode{},replyText:{},"
                    + "exchange:{},routingKey{}", cacheKey, replyCode, replyText, exchange, routingKey);
            RabbitMetaMessage metaMessage = redisTemplate.getMapField(MQConstants.MQ_PRODUCER_RETRY_KEY, cacheKey, RabbitMetaMessage.class);
            metaMessage.setReturnCallback(true);

            // 由于amqp奇葩协议规定，return比confirm先回调
            redisTemplate.addMap(MQConstants.MQ_PRODUCER_RETRY_KEY, cacheKey, metaMessage);
        });
        return rabbitTemplate;
    }

    /**
     * 解决由于消费异常导致，一致抛异常问题
     * {@link https://yemengying.com/2017/01/03/rabbitmq-classMapper/} 参照这个博文，说我会出现发送消息的类和接受消息的类必须是一样的，不仅是要里面的字段一样，类名一样，连类的包路径都要一样。
     * todo 暂时还未验证
     * 所以当系统1使用 JsonMessageConverter 发送消息类A给系统2时，系统2可以有如下几种方式来接收：
     * 1.依赖系统1的jar包,直接使用类A来接收
     * 2.不依赖系统1的jar包，自己建一个和A一模一样的类，连名称，包路径都一样
     * 3.负责监听 queue 的类实现 MessageListener 接口，直接接收 Message 类，再自己转换(现在项目用的是这种方式)
     *{@link DefaultJackson2JsonMessageConverter}  解决这种问题，在消费端监听容器中配置转换的type即可，
     *  {@link BusQueueConsumerConfig}用法参照单元测试
     *
     * @author:
     * @date :2018/1/11
     */
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        Jackson2JsonMessageConverter jsonMessageConverter = new Jackson2JsonMessageConverter();
        //Jackson2JsonMessageConverter jsonMessageConverter = new Jackson2JsonMessageConverter(jsonMapper());
        return jsonMessageConverter;
    }


    @Bean
    public DefaultJackson2JsonMessageConverter jackson2JsonMessageConverter() {
        DefaultJackson2JsonMessageConverter jsonMessageConverter = new DefaultJackson2JsonMessageConverter();
        return jsonMessageConverter;
    }
    @Bean
    public Gson2JsonMessageConverter gson2JsonMessageConverter() {
        Gson2JsonMessageConverter jsonMessageConverter = new Gson2JsonMessageConverter();
        return jsonMessageConverter;
    }

    @Bean
    public JsonMapper jsonMapper() {
        JsonMapper jsonMapper = new JsonMapper();
        return jsonMapper;
    }
}
